package com.alen.websocket

import android.os.Handler
import android.os.Looper
import com.alen.websocket.util.LogUtil
import com.alen.websocket.util.NetworkObserver
import okhttp3.*
import okio.ByteString
import java.util.concurrent.ExecutorService
import java.util.concurrent.Executors
import java.util.concurrent.LinkedBlockingDeque
import java.util.concurrent.TimeUnit

class WebSocketClient {

    private var config: Config = Config.Builder().build()

    private val queue = LinkedBlockingDeque<Message>(32)
    private var executorService: ExecutorService? = null

    private var url: String = ""
    private var client: OkHttpClient? = null
    private val listeners = mutableListOf<SocketCallback>()

    private var reconnectCount = 0

    @Volatile
    var state: WebSocketKit.SocketState = WebSocketKit.SocketState.Uninit

    var webSocket: WebSocket? = null

    fun setUrl(url: String): WebSocketClient {
        this.url = url
        return this
    }

    fun setConfig(config: Config): WebSocketClient {
        this.config = config
        return this
    }

    fun setOkHttp(client: OkHttpClient): WebSocketClient {
        this.client = client
        return this
    }

    private val mainHandle = Handler(Looper.getMainLooper())
    private val reconnectTask = Runnable {
        reconnect()
        checkReconnect()
    }

    fun startConnect() {
        if (url.isNullOrEmpty()) {
            LogUtil.e(config.logTag, "=========startConnect========url is null or empty!")
        } else {
            queue.clear()
            createSocket()
            checkReconnect()
        }
    }

    fun isConnect(): Boolean {
        return state == WebSocketKit.SocketState.Open
    }

    fun reconnect() {
        if ((state == WebSocketKit.SocketState.Closed || state == WebSocketKit.SocketState.Closing)
                && NetworkObserver.isNetAvailable(WebSocketKit.context)
        ) {
            doReconnect()
        }
    }

    private fun doReconnect() {
        LogUtil.e(config.logTag, "=========reconnect========startConnect")
        handleReconnect()
        createSocket()
    }

    private fun checkReconnect() {
        mainHandle.removeCallbacks(reconnectTask)
        if (config.autoReconnect) {
            mainHandle.postDelayed(
                    reconnectTask,
                    config.reconnectIntervalTimeUnit.toMillis(config.reconnectInterval)
            )
        }
    }


    private val clientListener: WebSocketListener = object : WebSocketListener() {
        override fun onOpen(webSocket: WebSocket, response: Response) {
            LogUtil.d(config.logTag, "=========onOpen========")
            state = WebSocketKit.SocketState.Open
            handleOpen()
        }

        override fun onMessage(webSocket: WebSocket, text: String) {
            LogUtil.d(config.logTag, "=========onMessage========message=$text")
            handleReceive(text)
        }

        override fun onMessage(webSocket: WebSocket, bytes: ByteString) {
            LogUtil.d(config.logTag, "=========onMessage========message=${bytes.base64()}")
            handleReceive(bytes)
        }

        override fun onClosing(webSocket: WebSocket, code: Int, reason: String) {
            LogUtil.d(config.logTag, "=========onClosing====code=$code====reason=${reason}")
            state = WebSocketKit.SocketState.Closing
            webSocket.close(1002, "on closing")
//            handleClose(reason)
        }

        override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
            LogUtil.d(config.logTag, "=========onClosed====code=$code====reason=${reason}")
            state = WebSocketKit.SocketState.Closed
            handleClose(reason)
        }

        override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
            LogUtil.d(
                    config.logTag,
                    "=========onFailure====url=${webSocket.request().url.toUrl()}>>>>> reason=${t.message}"
            )
            state = WebSocketKit.SocketState.Closed
            handleClose(t.message ?: "")
        }
    }

    private fun createSocket() {
        if (client == null) {
            createClient()
        }
        webSocket = client!!.newWebSocket(createRequest(), clientListener)
    }

    private fun createClient() {
        val builder = OkHttpClient.Builder()
        if (config.sslSocketFactory != null && config.trustManager != null) {
            builder.sslSocketFactory(config.sslSocketFactory, config.trustManager)
        }
        if (config.enablePing) {
            builder.pingInterval(config.pingInterval, TimeUnit.SECONDS)
        }
        client = builder.build()
    }

    private fun createRequest(): Request {
        if (url.isNullOrEmpty()) throw NullPointerException("url can't be null or empty")

        return Request.Builder().url(url).build()
    }

    fun send(message: ByteString) {
        sendMessage(ByteStringMessage(message))
    }

    fun send(message: String) {
        sendMessage(StringMessage(message))
    }

    fun sendMessage(message: Message) {
        if (queue.remainingCapacity() < 1) {
            queue.removeFirst()
        }
        queue.add(message)
    }


    fun close(immediate: Boolean = false) {
        if (isConnect()) {
            if (immediate) {
                queue.clear()
                mainHandle.removeCallbacks(reconnectTask)
                webSocket?.close(1001, "client requests close!")
                sendMessage(CloseMessage())
            } else {
                sendMessage(CloseMessage())
            }
        } else {
            queue.clear()
            mainHandle.removeCallbacks(reconnectTask)
        }
    }

    private fun takeMessageSend() {
//        if (executorService != null) {
//            queue.clear()
//            executorService?.shutdownNow()
//        }
        if (executorService != null && !executorService!!.isTerminated && !executorService!!.isShutdown)
            return

        executorService = Executors.newSingleThreadExecutor()
        executorService?.execute {
            while (true) {
                try {
                    var message: Message = queue.take()
                    when (message) {
                        is CloseMessage -> {
                            LogUtil.d(
                                    config.logTag, "=========queue===${queue.size}===CloseMessage"
                            )
                            close(true)
                            executorService = null
                            return@execute
                        }
                        is StringMessage -> {
                            LogUtil.d(
                                    config.logTag, "=========queue===${queue.size}===TextMessage=${message.content}"
                            )
                            webSocket?.send(message.content)
                        }
                        is ByteStringMessage -> {
                            LogUtil.d(
                                    config.logTag, "=========queue===${queue.size}===ByteMessage=${message.content}"
                            )
                            webSocket?.send(message.content)
                        }
                        is CustomMessage -> {
                            val customMessage = message.content()
                            LogUtil.d(
                                    config.logTag, "=========queue===${queue.size}===CustomMessage=${customMessage}"
                            )
                            webSocket?.send(customMessage)
                        }
                        else -> {
                            LogUtil.d(
                                    config.logTag, "=========takeMessageSend====$message can't be send"
                            )
                        }
                    }
                } catch (ex: InterruptedException) {
                    ex.printStackTrace()
                } catch (ex: Exception) {
                    ex.printStackTrace()
                }
            }

        }
    }

    private fun handleOpen() {
        reconnectCount = 0
        takeMessageSend()
        mainHandle.post {
            listeners.forEach {
                it.onOpen()
            }
        }
    }

    private fun handleReconnect() {
        mainHandle.post {
            listeners.forEach {
                it.onReconnect()
            }
        }
    }

    private fun handleClose(reason: String, needReconnect: Boolean = false) {
//        if (executorService?.isShutdown == false || executorService?.isTerminated == false) {
//            sendMessage(CloseMessage())
//        }
        mainHandle.post {
            listeners.forEach {
                it.onClose(reason)
            }
        }
        if (needReconnect)
            mainHandle.post { reconnect() }
    }

    private fun handleReceive(message: String) {
        mainHandle.post {
            listeners.forEach {
                it.onMessage(message)
            }
        }
    }

    private fun handleReceive(bytes: ByteString) {
        mainHandle.post {
            listeners.forEach {
                it.onMessage(bytes)
            }
        }
    }


    fun addListener(webSocketListener: SocketCallback) {
        if (!listeners.contains(webSocketListener)) {
            listeners.add(webSocketListener)
        }
    }

    fun removeListener(webSocketListener: SocketCallback) {
        if (listeners.contains(webSocketListener)) {
            listeners.remove(webSocketListener)
        }
    }

    fun clearListener() {
        listeners.clear()
    }

}